package io.kiki.sba.registry.server.shared.providedata;

import com.google.common.collect.Lists;
import io.kiki.sba.registry.common.model.metaserver.ProvideData;
import io.kiki.sba.registry.server.shared.meta.MetaServerService;
import io.kiki.sba.registry.util.ConcurrentUtils;
import io.kiki.sba.registry.util.WakeUpLoopExecuteTask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

public abstract class AbstractProvideDataWatcher {
    private static final Logger logger = LoggerFactory.getLogger(AbstractProvideDataWatcher.class);
    protected final WatcherDog watcherDog = new WatcherDog();
    protected final String name;
    protected final Thread thread;
    private final Map<String, DataWatcher> watcherMap = new ConcurrentHashMap<>(1024);
    @Autowired
    protected MetaServerService metaNodeService;

    protected AbstractProvideDataWatcher(String name) {
        this.name = name;
        thread = ConcurrentUtils.createDaemonThread("provide-data-watcher-" + name, watcherDog);
    }

    static List<Map<String, Long>> transferQuery(Collection<DataWatcher> dataWatchers, int batchSize) {
        if (dataWatchers.isEmpty()) {
            return Collections.emptyList();
        }
        List<List<DataWatcher>> partitioned = Lists.partition(new ArrayList<>(dataWatchers), batchSize);
        List<Map<String, Long>> mapList = new ArrayList<>(partitioned.size());
        for (List<DataWatcher> list : partitioned) {
            Map<String, Long> map = new HashMap<>(list.size());
            for (DataWatcher dataWatcher : list) {
                map.put(dataWatcher.dataInfoId, dataWatcher.provideData == null ? -1 : dataWatcher.provideData.getVersion());
            }
            mapList.add(map);
        }
        return mapList;
    }

    public void start() {
        thread.start();
    }

    public boolean watch(String dataInfoId) {
        DataWatcher dataWatcher = watcherMap.get(dataInfoId);
        if (dataWatcher != null) {
            dataWatcher.watchTimestamp = System.currentTimeMillis();
            return true;
        }
        if (watcherMap.putIfAbsent(dataInfoId, new DataWatcher(dataInfoId)) == null) {
            watcherDog.wakeup();
            return true;
        }
        return false;
    }

    public int refreshWatch(Collection<String> watchers) {
        int count = 0;
        for (String dataID : watchers) {
            if (watch(dataID)) {
                count++;
            }
        }
        return count;
    }

    void clean() {
        final long expireTs = System.currentTimeMillis() - watcherLeaseSecs() * 1000;
        for (DataWatcher dataWatcher : watcherMap.values()) {
            if (dataWatcher.watchTimestamp < expireTs) {
                watcherMap.remove(dataWatcher.dataInfoId);
                logger.info("clean watcher {}, watchTs", dataWatcher.dataInfoId, new Date(dataWatcher.watchTimestamp));
            }
        }
    }

    void fetch() {
        List<Map<String, Long>> queryList = transferQuery(watcherMap.values(), fetchBatchSize());
        for (Map<String, Long> query : queryList) {
            try {
                Map<String, ProvideData> provideDataMap = metaNodeService.getProvideData(query);
                updateFetchData(provideDataMap);
            } catch (Throwable e) {
                logger.error("failed to fetch data:{}", query.keySet(), e);
            }
        }
    }

    void updateFetchData(Map<String, ProvideData> provideDataMap) {
        for (Map.Entry<String, ProvideData> entry : provideDataMap.entrySet()) {
            final DataWatcher dataWatcher = watcherMap.get(entry.getKey());
            if (dataWatcher == null) {
                logger.info("DataWatcher not found when update: {}", entry.getKey());
                continue;
            }
            ProvideData provideData = entry.getValue();
            if (dataWatcher.provideData == null) {
                dataWatcher.provideData = entry.getValue();
                logger.info("DataWatcher init provideData: {}, {}", entry.getKey(), provideData);
                continue;
            }
            if (dataWatcher.provideData.getVersion() >= provideData.getVersion()) {
                logger.warn("DataWatcher skip provideData: {}, current={}, fetch={}", entry.getKey(), dataWatcher.provideData.getVersion(), provideData.getVersion());
                continue;
            }
            dataWatcher.provideData = provideData;
            logger.info("DataWatcher update provideData: {}, {}", entry.getKey(), provideData);
        }
    }

    public ProvideData get(String dataInfoId) {
        DataWatcher watcher = watcherMap.get(dataInfoId);
        return watcher != null ? watcher.provideData : null;
    }

    protected abstract int fetchBatchSize();

    protected abstract int fetchIntervalMillis();

    protected abstract int watcherLeaseSecs();


    DataWatcher getDataWatcher(String dataInfoId) {
        return watcherMap.get(dataInfoId);
    }

    static final class DataWatcher {
        final String dataInfoId;
        volatile long watchTimestamp = System.currentTimeMillis();
        volatile ProvideData provideData;

        DataWatcher(String dataInfoId) {
            this.dataInfoId = dataInfoId;
        }
    }

    protected final class WatcherDog extends WakeUpLoopExecuteTask {

        @Override
        public void _execute_() {
            clean();
            // TODO need to avoid fetch too frequently by wakeup
            fetch();
        }

        @Override
        public int getWaitingMillis() {
            return fetchIntervalMillis();
        }
    }
}
